第三步结构化输出优化

This commit is contained in:
fengxiangdebaozi committed 2026-06-17 21:31:22 +08:00
1 parent fca39eeaee
commit d3880f51b3
10 files changed
+1095 -127

No files matched your search

@@ -0,0 +1,106 @@
/**
* Step3 流程图「后端抽取 + 校验」SSE 客户端。
*
* 对应后端 `POST /api/flow-extract/stream`:后端独立调模型(不走总结线程,避开 coordinatorPrompt
* 的报告锚定)两段式产图——①流式抽取 flow-json(实时吐回,前端原样显示给用户看)→ ②模型校验/修复
* → 返回校验后的结构化图数据 `{title, nodes, edges}`。
*
* 这样把不稳定的「文本→结构」抽取 + 校验放后端,前端只负责实时显示抽取过程 + 拿最终图绘制。
*/
import { apiFetch } from "@/core/api/fetch-client";
import { getBackendBaseURL } from "@/core/config";
import type { FlowGraphData } from "@/roundtable-planning/lib/flow-graph";
export interface FlowExtractRequest {
/** 报告 Markdown(优先数据来源)。 */
reportMd?: string | null;
/** 各席位交付(无报告时的数据来源)。 */
seatDeliveries?: Array<{ sender: string; content: string }>;
/** 业务链层级(stages 的 goal)。 */
chainLevels?: string[];
/** 业务链总控编排提示(写明层级时优先用它)。 */
coordinatorPrompt?: string | null;
/** 推理模型名(沿用第三步所选)。 */
model?: string;
/** Phase 2 问答修改:当前流程图 + 指令 → 按指令改图(其余不变)。 */
currentGraph?: FlowGraphData | null;
instruction?: string;
signal?: AbortSignal;
/** 抽取流式增量(含 `<think>` 思考 + 正文 + JSON,前端原样渲染:思考块 + 代码块)。 */
onExtractDelta?: (text: string) => void;
/** 抽取结束、进入校验阶段。 */
onValidating?: () => void;
/** 校验/修复流式增量(同样含思考 + 正文,前端实时显示校验过程)。 */
onValidateDelta?: (text: string) => void;
}
/**
* 消费 SSE,返回校验后的最终图。抽取阶段通过 `onExtractDelta` 实时回吐;
* 出错(后端 error 帧 / 网络)抛异常。
*/
export async function streamFlowExtract(req: FlowExtractRequest): Promise<FlowGraphData> {
const res = await apiFetch(`${getBackendBaseURL()}/api/flow-extract/stream`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
report_md: req.reportMd ?? null,
seat_deliveries: (req.seatDeliveries ?? []).map((s) => ({ sender: s.sender, content: s.content })),
chain_levels: req.chainLevels ?? [],
coordinator_prompt: req.coordinatorPrompt ?? null,
model: req.model,
current_graph: req.currentGraph ?? null,
instruction: req.instruction ?? null,
}),
signal: req.signal,
});
if (!res.ok || !res.body) {
const text = await res.text().catch(() => "");
throw new Error(`flow-extract failed (${res.status}): ${text}`);
}
const reader = res.body.getReader();
const decoder = new TextDecoder();
let buffer = "";
let graph: FlowGraphData | null = null;
let errored: string | null = null;
while (true) {
const { done, value } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
const lines = buffer.split("\n");
buffer = lines.pop() ?? "";
for (const rawLine of lines) {
const line = rawLine.trimEnd();
if (!line.startsWith("data:")) continue;
const data = line.slice(5).trim();
if (!data) continue;
let frame: Record<string, unknown>;
try {
frame = JSON.parse(data) as Record<string, unknown>;
} catch {
continue;
}
const event = typeof frame.event === "string" ? frame.event : "";
if (event === "extract_delta") {
const c = typeof frame.content === "string" ? frame.content : "";
if (c) req.onExtractDelta?.(c);
} else if (event === "validating") {
req.onValidating?.();
} else if (event === "validate_delta") {
const c = typeof frame.content === "string" ? frame.content : "";
if (c) req.onValidateDelta?.(c);
} else if (event === "final") {
graph = (frame.graph as FlowGraphData) ?? null;
} else if (event === "error") {
errored = typeof frame.detail === "string" ? frame.detail : "未知错误";
}
}
}
if (errored) throw new Error(errored);
if (!graph || !Array.isArray(graph.nodes) || graph.nodes.length === 0) {
throw new Error("flow-extract 未返回有效流程图");
}
return graph;
}
@@ -15,6 +15,7 @@ import {
useRef,
useState,
} from "react";
import { createPortal } from "react-dom";
import {
applyNodeChanges,
Background,
@@ -32,11 +33,14 @@ import {
} from "@xyflow/react";
import "@xyflow/react/dist/style.css";
import {
Check,
FileText,
Loader2,
LocateFixed,
Maximize2,
Minimize2,
Pencil,
Trash2,
Workflow,
X,
ZoomIn,
@@ -45,6 +49,8 @@ import {
import { cn } from "@/lib/utils";
import {
applyNodeEdit,
deleteNodeFromGraph,
FLOW_NODE_TYPE_META,
FLOW_NODE_W,
layoutFlowGraph,
@@ -59,6 +65,8 @@ interface FlowGraphPanelProps {
onClose: () => void;
/** 沙箱里有 md 报告时提供:工具栏显示「MD 报告」按钮,点击切回 Markdown 预览。 */
onSwitchToReport?: () => void;
/** Phase 3:画布手动编辑——传入则节点详情弹窗出现「编辑 / 删除」,回写整张新图。 */
onChange?: (next: FlowGraphData) => void;
}
// ── 自定义节点(按 type 着色的卡片)────────────────────────────────────────
@@ -106,6 +114,7 @@ function FlowCanvasInner({
onToggleFullscreen,
onClose,
onSwitchToReport,
onChange,
}: FlowGraphPanelProps & { fullscreen: boolean; onToggleFullscreen: () => void }) {
const rf = useReactFlow();
const [selectedId, setSelectedId] = useState<string | null>(null);
@@ -276,13 +285,14 @@ function FlowCanvasInner({
</div>
</div>
{/* 节点详情弹窗 */}
{/* 节点详情弹窗(传入 onChange 时可编辑 / 删除节点) */}
{selectedNode && (
<NodeDetailModal
node={selectedNode}
data={data}
onClose={() => setSelectedId(null)}
onJump={(id) => setSelectedId(id)}
onChange={onChange}
/>
)}
</div>
@@ -317,17 +327,38 @@ function NodeDetailModal({
data,
onClose,
onJump,
onChange,
}: {
node: FlowGraphNode;
data: FlowGraphData;
onClose: () => void;
onJump: (id: string) => void;
/** Phase 3:传入则可编辑 / 删除该节点(回写整张新图)。 */
onChange?: (next: FlowGraphData) => void;
}) {
const meta = FLOW_NODE_TYPE_META[node.type];
const labelOf = (id: string) => data.nodes.find((n) => n.id === id)?.label ?? id;
const incoming = data.edges.filter((e) => e.to === node.id);
const outgoing = data.edges.filter((e) => e.from === node.id);
const [editing, setEditing] = useState(false);
const [label, setLabel] = useState(node.label);
const [group, setGroup] = useState(node.group ?? "");
const [detail, setDetail] = useState(node.detail ?? "");
const handleSave = () => {
onChange?.(applyNodeEdit(data, node.id, { label, group, detail }));
setEditing(false);
onClose();
};
const handleDelete = () => {
if (typeof window !== "undefined" && !window.confirm(`删除节点「${node.label}」及其连线?此操作不可撤销。`)) {
return;
}
onChange?.(deleteNodeFromGraph(data, node.id));
onClose();
};
return (
<div
className="absolute inset-0 z-30 flex items-center justify-center bg-black/30 p-4 backdrop-blur-[1px]"
@@ -361,27 +392,110 @@ function NodeDetailModal({
<X className="h-4 w-4" />
</button>
</div>
<div className="max-h-[50vh] space-y-3 overflow-y-auto p-4">
<div>
<div className="mb-1 text-[10px] font-bold uppercase tracking-widest text-slate-400">
详细说明
</div>
<p className="text-xs leading-relaxed text-slate-700">
{node.detail?.trim() || "(该节点未提供详细说明)"}
</p>
{editing ? (
<div className="max-h-[55vh] space-y-3 overflow-y-auto p-4">
<EditField label="节点名称(label)" value={label} onChange={setLabel} />
<EditField label="所属层级(group,如 目的 / 行为体)" value={group} onChange={setGroup} />
<EditField label="详细内容(detail,名称+理由)" value={detail} onChange={setDetail} multiline />
</div>
{incoming.length > 0 && (
<ConnectionList title="上游(来自)" entries={incoming.map((e) => ({ id: e.from, label: labelOf(e.from), edgeLabel: e.label }))} onJump={onJump} />
)}
{outgoing.length > 0 && (
<ConnectionList title="下游(去往)" entries={outgoing.map((e) => ({ id: e.to, label: labelOf(e.to), edgeLabel: e.label }))} onJump={onJump} />
)}
</div>
) : (
<div className="max-h-[50vh] space-y-3 overflow-y-auto p-4">
<div>
<div className="mb-1 text-[10px] font-bold uppercase tracking-widest text-slate-400">详细说明</div>
<p className="text-xs leading-relaxed text-slate-700 whitespace-pre-wrap">
{node.detail?.trim() || "(该节点未提供详细说明)"}
</p>
</div>
{incoming.length > 0 && (
<ConnectionList title="上游(来自)" entries={incoming.map((e) => ({ id: e.from, label: labelOf(e.from), edgeLabel: e.label }))} onJump={onJump} />
)}
{outgoing.length > 0 && (
<ConnectionList title="下游(去往)" entries={outgoing.map((e) => ({ id: e.to, label: labelOf(e.to), edgeLabel: e.label }))} onJump={onJump} />
)}
</div>
)}
{/* 编辑 / 删除 操作栏(仅传入 onChange 时显示) */}
{onChange && (
<div className="flex items-center justify-between gap-2 border-t border-slate-100 bg-slate-50/60 px-4 py-2.5">
{editing ? (
<>
<button
type="button"
onClick={() => setEditing(false)}
className="rounded-lg px-2.5 py-1.5 text-[11px] font-bold text-slate-500 hover:bg-slate-200/60"
>
取消
</button>
<button
type="button"
onClick={handleSave}
className="inline-flex items-center gap-1 rounded-lg bg-primary px-3 py-1.5 text-[11px] font-bold text-white hover:bg-[var(--brand-900)]"
>
<Check className="h-3.5 w-3.5" />
保存
</button>
</>
) : (
<>
<button
type="button"
onClick={handleDelete}
className="inline-flex items-center gap-1 rounded-lg border border-rose-200 px-2.5 py-1.5 text-[11px] font-bold text-rose-600 hover:bg-rose-50"
>
<Trash2 className="h-3.5 w-3.5" />
删除节点
</button>
<button
type="button"
onClick={() => setEditing(true)}
className="inline-flex items-center gap-1 rounded-lg border border-slate-200 px-3 py-1.5 text-[11px] font-bold text-slate-700 hover:border-primary/40 hover:text-primary"
>
<Pencil className="h-3.5 w-3.5" />
编辑
</button>
</>
)}
</div>
)}
</div>
</div>
);
}
function EditField({
label,
value,
onChange,
multiline,
}: {
label: string;
value: string;
onChange: (v: string) => void;
multiline?: boolean;
}) {
return (
<div>
<div className="mb-1 text-[10px] font-bold uppercase tracking-widest text-slate-400">{label}</div>
{multiline ? (
<textarea
value={value}
onChange={(e) => onChange(e.target.value)}
rows={4}
className="w-full resize-none rounded-lg border border-slate-200 bg-white px-2.5 py-1.5 text-xs leading-relaxed text-slate-800 focus:border-primary focus:outline-none focus:ring-1 focus:ring-[var(--brand-200)]/50"
/>
) : (
<input
value={value}
onChange={(e) => onChange(e.target.value)}
className="w-full rounded-lg border border-slate-200 bg-white px-2.5 py-1.5 text-xs text-slate-800 focus:border-primary focus:outline-none focus:ring-1 focus:ring-[var(--brand-200)]/50"
/>
)}
</div>
);
}
function ConnectionList({
title,
entries,
@@ -420,14 +534,15 @@ function ConnectionList({
// ── 外壳:覆盖右栏 / 全屏切换 ───────────────────────────────────────────────
export function FlowGraphPanel({ data, streaming, onClose, onSwitchToReport }: FlowGraphPanelProps) {
export function FlowGraphPanel({ data, streaming, onClose, onSwitchToReport, onChange }: FlowGraphPanelProps) {
const [fullscreen, setFullscreen] = useState(false);
return (
const inner = (
<div
className={cn(
"flex flex-col bg-white",
// 非全屏=填满所在面板(Step3 放在右侧沙箱位,不覆盖左侧消息列表);全屏才整页覆盖。
fullscreen ? "fixed inset-0 z-[80]" : "relative h-full min-h-0 w-full",
// 非全屏=填满所在面板(Step3 放在右侧沙箱位,不覆盖左侧消息列表);
// 全屏=覆盖整个视口(z 高于 app 顶部导航,保证顶部工具栏可见、可点「退出全屏」)。
fullscreen ? "fixed inset-0 z-[9999]" : "relative h-full min-h-0 w-full",
)}
>
<ReactFlowProvider>
@@ -438,8 +553,15 @@ export function FlowGraphPanel({ data, streaming, onClose, onSwitchToReport }: F
onToggleFullscreen={() => setFullscreen((f) => !f)}
onClose={onClose}
onSwitchToReport={onSwitchToReport}
onChange={onChange}
/>
</ReactFlowProvider>
</div>
);
// 全屏时 portal 到 body:脱离任何带 transform/overflow 的祖先(Step2 沙箱分栏等),
// 否则 `fixed` 会被祖先裁剪、顶部工具栏被 app 导航栏盖住(即本次 bug)。
if (fullscreen && typeof document !== "undefined") {
return createPortal(inner, document.body);
}
return inner;
}
@@ -46,7 +46,6 @@ import { FlowGraphPanel } from "@/roundtable-planning/components/FlowGraphPanel"
import {
detectFlowIntent,
extractFlowGraph,
stripFlowFence,
FLOW_FENCE_OPEN,
} from "@/roundtable-planning/lib/flow-graph";
import { ScrollToBottomButton } from "@/roundtable-planning/components/ScrollToBottomButton";
@@ -139,9 +138,18 @@ export function Step3SummaryMode({
const { open: artifactsOpen } = useArtifacts();
// ── 结构化流程图:流式提取回复末尾的 flow-json,增量绘制 ──────────────────
// 取最后一条含 flow-json 的智能体气泡(多轮修改后以最新一份为准)。
// ── 结构化流程图(Phase 1:后端「抽取+校验」)────────────────────────────
// 数据来源优先级:
// · flowPhase==="done" → 用后端校验后的结构化图 summary.flowGraph 绘制;
// · 抽取/校验进行中 → 不画(用户看对话气泡里实时滚动的 JSON + 状态条),等校验完成再绘制;
// · idle/error → 回退到对话气泡里解析 flow-json(手动 flow 续问路径 / 旧逻辑兜底)。
const flowGraph = useMemo(() => {
if (summary.flowPhase === "done" && summary.flowGraph) {
return { graph: summary.flowGraph, streaming: false };
}
if (summary.flowPhase === "extracting" || summary.flowPhase === "validating") {
return null;
}
for (let i = dialogues.length - 1; i >= 0; i--) {
const d = dialogues[i];
if (d.avatarType === "human") continue;
@@ -151,7 +159,7 @@ export function Step3SummaryMode({
if (graph) return { graph, streaming: d.streaming === true };
}
return null;
}, [dialogues]);
}, [dialogues, summary.flowPhase, summary.flowGraph]);
const [flowView, setFlowView] = useState(false);
const flowAutoOpenedRef = useRef(false);
@@ -196,16 +204,10 @@ export function Step3SummaryMode({
ask(message, files);
};
// 气泡里不展示生 JSON:把 flow-json 块替换成一句引导语。
const displayDialogues = useMemo(
() =>
dialogues.map((d) => {
const content = d.content ?? "";
if (!content.includes(FLOW_FENCE_OPEN)) return d;
return { ...d, content: stripFlowFence(content, d.streaming === true) };
}),
[dialogues],
);
// 流程图抽取/校验的过程(思考 `<think>` + 文字说明 + flow-json 代码块)全部原样交给
// Step2Message 渲染:它用 splitInlineReasoning 把思考折叠成思考块、其余按 markdown 渲染
// (flow-json 显示为代码块)。用户要求显示完整过程、不只显示代码块,故不再 strip。
const displayDialogues = dialogues;
const [leftCollapsed, setLeftCollapsed] = useState(false);
// 右侧面板(沙箱预览或流程图画布)露出时自动收起左栏,关闭时复原——边沿触发,手动切换不回弹。
const rightPanelVisible = artifactsOpen || (!!flowGraph && flowView);
@@ -397,6 +399,7 @@ export function Step3SummaryMode({
}
: undefined
}
onChange={summary.setFlowGraph}
/>
) : undefined
}
@@ -407,6 +410,8 @@ export function Step3SummaryMode({
error={error}
busy={busy}
mdAvailable={!!mdArtifact}
flowPhase={summary.flowPhase}
onRegenerateFlow={summary.regenerateFlow}
onRetry={redraw}
onAsk={handleAsk}
onPause={pause}
@@ -549,6 +554,8 @@ function AgentProgressPanel({
error,
busy,
mdAvailable,
flowPhase,
onRegenerateFlow,
onRetry,
onAsk,
onPause,
@@ -565,6 +572,10 @@ function AgentProgressPanel({
busy: boolean;
/** 沙箱里是否已有 md 报告 —— 决定提示语(纯流程图轮没有 .md 可预览/下载)。 */
mdAvailable: boolean;
/** 后端流程图抽取阶段(Phase 1)。 */
flowPhase: UseStep3SummaryReturn["flowPhase"];
/** 仅重新生成流程图(模型挂了切模型后重试)。 */
onRegenerateFlow: () => void;
onRetry: () => void;
onAsk: (message: string, files?: File[]) => void;
onPause: () => void;
@@ -705,6 +716,28 @@ function AgentProgressPanel({
</div>
)}
{/* 后端流程图「抽取+校验」状态条(Phase 1)。 */}
{(flowPhase === "extracting" || flowPhase === "validating") && (
<div className="inline-flex items-center gap-2 rounded-full border border-blue-200 bg-blue-50 px-2.5 py-1 text-[11px] font-bold text-blue-600">
<Loader2 className="w-3 h-3 animate-spin" />
{flowPhase === "extracting" ? "结构化流程图抽取中(JSON 实时输出)…" : "流程图校验 / 修复中…"}
</div>
)}
{flowPhase === "error" && (
<div className="rounded-xl border border-amber-200 bg-amber-50 p-3 text-xs text-amber-800">
<div className="font-bold mb-1">流程图生成失败</div>
<div className="leading-relaxed">模型可能不稳定。可在上方切换模型后重试。</div>
<button
type="button"
onClick={onRegenerateFlow}
className="mt-2 inline-flex items-center gap-1 text-[11px] font-bold text-amber-800 underline hover:text-amber-900"
>
<RefreshCw className="w-3 h-3" />
重新生成流程图
</button>
</div>
)}
{phase === "done" && (
<div className="flex items-center gap-2 text-xs font-bold text-emerald-600">
<CheckCircle className="w-4 h-4" />
@@ -40,7 +40,8 @@ import type {
Step2Dialogue,
StubAiMessage,
} from "@/roundtable-planning/hooks/useStep2Orchestration";
import { detectFlowIntent, hasFlowFence } from "@/roundtable-planning/lib/flow-graph";
import { detectFlowIntent, extractFlowGraph, hasFlowFence, serializeFlowGraph, type FlowGraphData } from "@/roundtable-planning/lib/flow-graph";
import { streamFlowExtract } from "@/roundtable-planning/api/flow-extract";
import { nowTime } from "@/roundtable-planning/lib/roundtable-constants";
/** 报告输出文件的固定虚拟路径(恢复历史快照时用它重建 stub)。 */
@@ -79,6 +80,14 @@ export interface UseStep3SummaryReturn {
error: string | null;
/** taskId/深链模式:md 报告生成后自动续跑流程图(组件据此让流程图画布自动聚焦)。 */
autoFlow: boolean;
/** 后端「抽取+校验」得到的校验后结构化流程图(Phase 1)。done 时据此绘制。 */
flowGraph: FlowGraphData | null;
/** 流程图生成阶段:idle / extracting(抽取流式) / validating(校验中) / done / error。 */
flowPhase: "idle" | "extracting" | "validating" | "done" | "error";
/** 仅重新生成流程图(复用当前 md/席位交付,不重跑报告)——模型挂了切模型后重试用。 */
regenerateFlow: () => void;
/** 手动编辑流程图(Phase 3:画布上改/删节点后回写 + 持久化)。 */
setFlowGraph: (next: FlowGraphData) => void;
summaryThreadId: string | null;
dialogues: Step2Dialogue[];
seatStubMessages: Record<string, StubAiMessage[]>;
@@ -220,61 +229,9 @@ const FLOW_STRUCTURE_RULE =
"⑥ **只画**这些层级实体节点;**不要**把『综合分析』『核心洞察』『分层传播策略 / 执行路线 / 建议』『结论』" +
"以及『阶段①②③④』等章节名 / 阶段名 / 汇总段落画成节点。";
/** 把各席位交付拼成一段内联材料(无报告时作流程图的数据来源)。 */
function inlineSeatDeliveries(seatDeliveries?: { sender: string; content: string }[]): string {
const list = (seatDeliveries ?? []).filter((s) => (s.content ?? "").trim());
if (list.length === 0) return "";
return list
.map((s, i) => `## 席位${i + 1}:${s.sender || "(未具名席位)"}\n${s.content.trim()}`)
.join("\n\n");
}
/**
* taskId/深链模式:md 报告生成完后自动续跑的「结构化输出」prompt。
*
* 数据基准(用户拍板):**有报告就基于报告**(报告已把各席位交付清晰总结过,直接提取实体、
* 不要再自己重新总结一遍);没有报告则基于**各席位交付产物**。两种来源都**整段内联**进 prompt
* (不依赖线程上下文),让模型逐条对着抽取节点。层级顺序优先用总控提示词里写明的层级,否则 chainLevels。
*/
function buildAutoFlowPrompt(opts: {
chainLevels?: string[];
coordinatorPrompt?: string | null;
reportMd?: string;
seatDeliveries?: { sender: string; content: string }[];
}): string {
const levels = (opts.chainLevels ?? []).map((s) => s.trim()).filter(Boolean);
const cp = opts.coordinatorPrompt?.trim();
const report = opts.reportMd?.trim() ?? "";
const seatMaterial = report ? "" : inlineSeatDeliveries(opts.seatDeliveries);
const parts: string[] = [];
if (report) {
parts.push(
"本轮只画一张流程图。**数据来源 = 下面这份你已写好的《方案总结报告》**(它已把各席位交付清晰总结过)——" +
"请**逐条对着报告抽取**已有的层级条目作为流程图节点,**不要再自己重新总结 / 概括 / 提炼**,也不要凭空新增报告里没有的内容。",
);
} else {
parts.push(
"本轮只画一张流程图。**数据来源 = 下面各席位的交付产物**——请**逐条对着这些交付抽取**已有的层级条目作为流程图节点,不要自己另起总结。",
);
}
parts.push(FLOW_ONLY_RULE);
// 整段内联数据来源(报告全文 / 席位交付),用分隔线包起来便于模型对照抽取。
if (report) {
parts.push("===== 方案总结报告全文(数据来源,逐条抽取,勿改写)=====\n" + report + "\n===== 报告结束 =====");
} else if (seatMaterial) {
parts.push("===== 各席位交付产物(数据来源,逐条抽取,勿改写)=====\n" + seatMaterial + "\n===== 交付结束 =====");
}
parts.push(FLOW_JSON_SHAPE);
parts.push(FLOW_STRUCTURE_RULE);
if (cp) {
parts.push("层级顺序优先采用「业务链总控编排提示」里写明的层级链(如它写的 任务 → 目的 → 行为体 → …)。");
} else if (levels.length > 0) {
parts.push(`层级顺序(自上而下):任务 → ${levels.join(" → ")}。`);
}
// 收尾再强调(首尾夹击,对抗线程里 coordinatorPrompt 的报告锚定)。
parts.push("再次强调:本轮不写任何 Markdown 报告正文 / 章节,整个回复除一句话说明外,只有一个完整的 ```flow-json 代码块。");
return parts.join("\n");
}
// 注:原 buildAutoFlowPrompt(在总结线程里续问吐 flow-json)已废弃 —— Phase 1 起自动结构化
// 输出改走后端独立端点 /api/flow-extract/stream(streamFlowExtract),抽取+校验在后端做。
// FLOW_ONLY_RULE / FLOW_STRUCTURE_RULE / FLOW_JSON_SHAPE 仍由手动 FLOW_ASK_INSTRUCTION 引用。
/** 首轮总结 prompt。`flowMode`=编排提示里明确要求了结构化输出/流程图 → 跳过 md 文件,直出流程图。 */
function buildSummaryPrompt(opts: {
@@ -452,6 +409,14 @@ export function useStep3Summary(
const [phase, setPhase] = useState<SummaryPhase>("idle");
const [error, setError] = useState<string | null>(null);
// 后端「抽取+校验」流程图(Phase 1):与 SummaryPhase 解耦(md 报告完成后才跑流程图)。
const [flowGraph, setFlowGraphState] = useState<FlowGraphData | null>(null);
const [flowPhase, setFlowPhase] = useState<"idle" | "extracting" | "validating" | "done" | "error">("idle");
// 镜像 flowGraph 供 ask()(问答修改)在 render 外读到最新图作 edit 入参。
const flowGraphRef = useRef<FlowGraphData | null>(null);
useEffect(() => {
flowGraphRef.current = flowGraph;
}, [flowGraph]);
const [summaryThreadId, setSummaryThreadId] = useState<string | null>(null);
const [dialogues, setDialogues] = useState<Step2Dialogue[]>([]);
const [seatStubMessages, setSeatStubMessages] = useState<
@@ -1010,6 +975,110 @@ export function useStep3Summary(
],
);
/**
* Phase 1:后端「抽取+校验」生成流程图。独立调 `/api/flow-extract/stream`(不进总结线程,
* 避开 coordinator_prompt 的报告锚定):抽取阶段把 JSON 流式吐进对话气泡(用户实时看),
* 校验完成后用校验后的结构化图绘制画布(done)。模型沿用第三步所选(selectedModel)。
*/
const runFlowExtract = useCallback(
async (
aborter: AbortController,
opts?: { instruction?: string; currentGraph?: FlowGraphData | null },
): Promise<void> => {
const isEdit = !!opts?.instruction && !!opts?.currentGraph;
setFlowPhase("extracting");
if (!isEdit) setFlowGraphState(null); // 编辑模式保留旧图直到新图就绪(防画布闪空)
// 两个气泡:抽取/修改 + 校验——把过程(思考 `<think>` + 文字 + flow-json 代码块)都留着展示,
// 由 Step2Message 渲染(思考折叠块 + markdown 代码块),不折叠成一句话。
const extractBubbleId = appendStreamingDialogue({
sender: isEdit ? "结构化流程图 · 修改" : "结构化流程图 · 抽取",
avatarType: "blue",
});
let validateBubbleId: number | null = null;
try {
const graph = await streamFlowExtract({
reportMd: currentMdRef.current,
seatDeliveries,
chainLevels,
coordinatorPrompt,
model: selectedModel || undefined,
currentGraph: opts?.currentGraph ?? null,
instruction: opts?.instruction,
signal: aborter.signal,
onExtractDelta: (t) => appendToDialogue(extractBubbleId, t),
onValidating: () => {
setFlowPhase("validating");
// 抽取气泡收尾(保留已流出的内容),另起校验气泡。
finalizeDialogue(extractBubbleId, "", "抽取完成");
validateBubbleId = appendStreamingDialogue({ sender: "结构化流程图 · 校验", avatarType: "blue" });
},
onValidateDelta: (t) => {
if (validateBubbleId != null) appendToDialogue(validateBubbleId, t);
},
});
if (aborter.signal.aborted) return;
setFlowGraphState(graph);
setFlowPhase("done");
flowActiveRef.current = true;
// 收尾两个气泡(保留过程内容)。校验后的图序列化进草稿快照 closing,恢复时重建画布。
if (validateBubbleId != null) finalizeDialogue(validateBubbleId, "", "流程图完成");
else finalizeDialogue(extractBubbleId, "", "流程图完成");
onSummarySavedRef.current?.({
md: currentMdRef.current,
threadId: liveThreadRef.current ?? "",
generatedAt: new Date().toISOString(),
model: selectedModel || "",
closing: serializeFlowGraph(graph),
});
} catch (err) {
if (aborter.signal.aborted) return;
const msg = err instanceof Error ? err.message : String(err);
setFlowPhase("error");
finalizeDialogue(
validateBubbleId ?? extractBubbleId,
`(流程图生成失败) ${msg}\n模型可能不稳定——可在上方切换模型后点「重新生成流程图」重试。`,
"流程图失败",
);
}
},
[
seatDeliveries,
chainLevels,
coordinatorPrompt,
selectedModel,
appendStreamingDialogue,
appendToDialogue,
finalizeDialogue,
],
);
/** 仅重新生成流程图(不重跑 md 报告):模型挂了切模型后重试 / 手动重画。 */
const regenerateFlow = useCallback(() => {
if (!currentMdRef.current.trim() && seatDeliveries.length === 0) return;
aborterRef.current?.abort();
const aborter = new AbortController();
aborterRef.current = aborter;
void runFlowExtract(aborter);
}, [runFlowExtract, seatDeliveries]);
/** Phase 3:画布手动改/删节点后回写图 + 持久化(序列化进草稿快照)。 */
const updateFlowGraph = useCallback(
(next: FlowGraphData) => {
setFlowGraphState(next);
setFlowPhase("done");
flowActiveRef.current = true;
const serialized = serializeFlowGraph(next);
onSummarySavedRef.current?.({
md: currentMdRef.current,
threadId: liveThreadRef.current ?? "",
generatedAt: new Date().toISOString(),
model: selectedModel || "",
closing: serialized,
});
},
[selectedModel],
);
const run = useCallback(async () => {
if (runningRef.current) return;
runningRef.current = true;
@@ -1021,6 +1090,8 @@ export function useStep3Summary(
currentMdRef.current = "";
liveThreadRef.current = null;
setSeatStubMessages({});
setFlowGraphState(null);
setFlowPhase("idle");
const aborter = new AbortController();
aborterRef.current = aborter;
@@ -1044,33 +1115,16 @@ export function useStep3Summary(
});
await streamOnce({ bubbleId, prompt, aborter, flowRound: flowMode });
// taskId/深链模式:报告生成完,**自动**续跑一轮结构化输出(流程图)。
// 数据来源:有报告则内联报告全文、否则内联各席位交付(buildAutoFlowPrompt 处理)。
// 故触发条件放宽为「有报告 **或** 有席位交付」——保证「无报告也能据席位交付出图」。
// 复用同一真线程,与手动「结构化输出」走同一渲染链路。
// taskId/深链模式:报告生成完,**自动**用后端「抽取+校验」端点生成流程图(Phase 1)。
// 不再在总结线程里续问吐 flow-json(那会被 coordinator_prompt 的报告锚定带偏、不稳);
// 改走独立端点:流式抽取(实时显示) → 校验修复 → 用校验后结构化图绘制。
// 有报告或有席位交付即触发(无报告也能据席位交付出图)。
if (
autoFlow &&
!aborter.signal.aborted &&
liveThreadRef.current &&
(currentMdRef.current.trim() || seatDeliveries.length > 0)
) {
const flowBubbleId = appendStreamingDialogue({
sender: "方案总结报告智能体",
avatarType: "blue",
});
const flowPrompt = buildAutoFlowPrompt({
chainLevels,
coordinatorPrompt,
reportMd: currentMdRef.current,
seatDeliveries,
});
await streamOnce({
bubbleId: flowBubbleId,
prompt: flowPrompt,
aborter,
seedThreadId: liveThreadRef.current,
flowRound: true,
});
await runFlowExtract(aborter);
}
} catch (err) {
if (aborter.signal.aborted) return;
@@ -1087,11 +1141,11 @@ export function useStep3Summary(
seatDeliveries,
consensusText,
coordinatorPrompt,
chainLevels,
autoFlow,
appendStreamingDialogue,
finalizeDialogue,
streamOnce,
runFlowExtract,
]);
const ask = useCallback(
@@ -1106,6 +1160,25 @@ export function useStep3Summary(
abort();
if (wasBusy) freezeStreaming("已打断");
// 流程图相关消息 → 走后端「抽取/编辑 + 校验」端点(流式显示过程 → 校验后绘制),
// 不再在总结线程吐 flow-json 做实时绘制。判定:
// · 已有流程图 且(显式 flow 关键词 或 上一轮主产物是流程图)→ **问答修改**(带当前图+指令,Phase 2);
// · 还没有流程图 但显式要流程图 → **首次生成**。
// 普通报告修改 / 问答(无 flow 上下文)仍走下方总结线程路径。
const flowIntentNow = detectFlowIntent(text);
const curGraph = flowGraphRef.current;
if (flowIntentNow || (curGraph && flowActiveRef.current)) {
appendUserDialogue(text);
const flowAborter = new AbortController();
aborterRef.current = flowAborter;
if (curGraph) {
await runFlowExtract(flowAborter, { instruction: text, currentGraph: curGraph });
} else {
await runFlowExtract(flowAborter);
}
return;
}
// 意图引导分三档(智能体按语义最终裁决,前端只给上下文,不做硬分类):
// ① 显式命中结构化/流程图关键词 → 直出 flow-json、不写 md;
// ② 没命中关键词但上一轮主产物是流程图 → 流程图上下文引导(笼统修改默认改图,太模糊则反问);
@@ -1219,6 +1292,7 @@ export function useStep3Summary(
appendUserDialogue,
appendStreamingDialogue,
streamOnce,
runFlowExtract,
],
);
@@ -1247,6 +1321,11 @@ export function useStep3Summary(
(snap: DraftStep3SummarySnapshot) => {
abort();
runningRef.current = false;
// 恢复流程图:快照 closing 里若存了 flow-json,解析成结构化图填进 flowGraph 状态,
// 让画布直接用它绘制(也供 Phase 3 手动编辑)。解析不出则回退到对话气泡解析。
const restoredGraph = snap.closing ? extractFlowGraph(snap.closing) : null;
setFlowGraphState(restoredGraph);
setFlowPhase(restoredGraph ? "done" : "idle");
const hasMd = snap.md.trim().length > 0;
const realThreadId = snap.threadId?.trim() || null;
const synthThreadId = realThreadId ?? "roundtable-summary-restored";
@@ -1330,6 +1409,10 @@ export function useStep3Summary(
phase,
error,
autoFlow: !!autoFlow,
flowGraph,
flowPhase,
regenerateFlow,
setFlowGraph: updateFlowGraph,
summaryThreadId,
dialogues,
seatStubMessages,
@@ -9,10 +9,13 @@ import assert from "node:assert/strict";
import test from "node:test";
import {
applyNodeEdit,
deleteNodeFromGraph,
detectFlowIntent,
extractFlowGraph,
hasFlowFence,
layoutFlowGraph,
serializeFlowGraph,
stripFlowFence,
} from "./flow-graph";
@@ -206,3 +209,51 @@ test("layoutFlowGraph:环不死循环", () => {
assert.equal(positioned.length, 2);
positioned.forEach((n) => assert.ok(Number.isFinite(n.x) && Number.isFinite(n.y)));
});
test("serializeFlowGraph:与 extractFlowGraph 往返一致(含 group/detail/边标签)", () => {
const text = [
"```flow-json",
'{"title":"方案","nodes":[',
'{"id":"t","label":"任务X","type":"start","group":"任务","detail":"目标说明"},',
'{"id":"p1","label":"目的1","type":"task","group":"目的","detail":"名称+理由","from":["t"]},',
'{"id":"d","label":"判断","type":"decision","from":[{"id":"p1","label":"可行"}]}',
"]}",
"```",
].join("\n");
const g = extractFlowGraph(text)!;
const round = extractFlowGraph(serializeFlowGraph(g))!;
assert.equal(round.title, "方案");
assert.deepEqual(
round.nodes.map((n) => [n.id, n.label, n.group, n.detail]),
g.nodes.map((n) => [n.id, n.label, n.group, n.detail]),
);
assert.deepEqual(round.edges, g.edges); // 含 {from:"p1",to:"d",label:"可行"}
});
test("applyNodeEdit:改 label/group/detail,其余节点不变", () => {
const g = extractFlowGraph(
'```flow-json\n{"nodes":[{"id":"a","label":"A","type":"task","group":"目的"},{"id":"b","label":"B","type":"role","group":"行为体","from":["a"]}]}\n```',
)!;
const next = applyNodeEdit(g, "a", { label: "新目的", group: "目标", detail: "改后的详情" });
const a = next.nodes.find((n) => n.id === "a")!;
assert.equal(a.label, "新目的");
assert.equal(a.group, "目标");
assert.equal(a.detail, "改后的详情");
// b 不变;边不变。
assert.equal(next.nodes.find((n) => n.id === "b")!.label, "B");
assert.deepEqual(next.edges, g.edges);
// 空 group/detail → 清空(置 undefined)。
const cleared = applyNodeEdit(g, "a", { group: " ", detail: "" });
const a2 = cleared.nodes.find((n) => n.id === "a")!;
assert.equal(a2.group, undefined);
});
test("deleteNodeFromGraph:删节点同时删掉与它相连的所有边", () => {
const g = extractFlowGraph(
'```flow-json\n{"nodes":[{"id":"t","label":"任务"},{"id":"p","label":"目的","from":["t"]},{"id":"b","label":"行为体","from":["p"]}]}\n```',
)!;
const next = deleteNodeFromGraph(g, "p");
assert.deepEqual(next.nodes.map((n) => n.id), ["t", "b"]);
// t→p 和 p→b 两条边都没了(b 变孤点)。
assert.equal(next.edges.length, 0);
});
@@ -281,6 +281,64 @@ export function extractFlowGraph(text: string): FlowGraphData | null {
return { title, nodes, edges: validEdges };
}
/**
* 把结构化图序列化回一个 ```flow-json 代码块(与 extractFlowGraph 互逆)。
* 用途:后端校验后拿到 FlowGraphData → 序列化存进草稿快照(closing),恢复时仍走
* extractFlowGraph 重建;手动编辑后也用它回存。`from` 由 edges 按 `to` 归集(带边标签)。
*/
export function serializeFlowGraph(data: FlowGraphData): string {
const fromMap = new Map<string, Array<string | { id: string; label: string }>>();
for (const e of data.edges) {
const list = fromMap.get(e.to) ?? [];
list.push(e.label ? { id: e.from, label: e.label } : e.from);
fromMap.set(e.to, list);
}
const nodes = data.nodes.map((n) => {
const o: Record<string, unknown> = { id: n.id, label: n.label, type: n.type };
if (n.group) o.group = n.group;
if (n.detail) o.detail = n.detail;
const from = fromMap.get(n.id);
if (from && from.length > 0) o.from = from;
return o;
});
return `${FLOW_FENCE_OPEN}\n${JSON.stringify({ title: data.title || "方案流程图", nodes })}\n\`\`\``;
}
/**
* Phase 3 画布手动编辑:改某节点的 label/group/detail(其余不变)。返回新图(不可变)。
*/
export function applyNodeEdit(
data: FlowGraphData,
id: string,
patch: { label?: string; group?: string; detail?: string },
): FlowGraphData {
return {
...data,
nodes: data.nodes.map((n) =>
n.id === id
? {
...n,
...(patch.label !== undefined ? { label: patch.label.trim() || n.label } : {}),
// group/detail 允许清空(传空串 → 置 undefined)。
...(patch.group !== undefined ? { group: patch.group.trim() || undefined } : {}),
...(patch.detail !== undefined ? { detail: patch.detail.trim() || undefined } : {}),
}
: n,
),
};
}
/**
* Phase 3 画布手动删除:删一个节点 + 所有与它相连的边。返回新图。
*/
export function deleteNodeFromGraph(data: FlowGraphData, id: string): FlowGraphData {
return {
...data,
nodes: data.nodes.filter((n) => n.id !== id),
edges: data.edges.filter((e) => e.from !== id && e.to !== id),
};
}
/**
* 识别文本(业务链编排提示 / 用户输入框消息)里是否明确要求了结构化输出 / 流程图。
* 命中 → 总结走「流程图模式」:跳过 .md 文件,报告用文字展示 + 末尾输出 flow-json。
@@ -1012,8 +1012,9 @@ export default function RoundtablePlanningPage() {
.filter((g) => g.length > 0),
[orchestrationPlan],
);
// taskId/深链模式:进第三步生成 md 后自动续跑结构化输出(流程图)。普通手动会商不自动。
const autoSummaryFlow = !!prefillRef.current?.taskId;
// 第三步首次生成报告后即**自动**续跑结构化输出(流程图)——不再限 taskId 深链,
// 任何会商进第三步首次出报告都自动出流程图(用户 2026-06-17 要求)。
const autoSummaryFlow = true;
const summaryFlow = useStep3Summary({
intentReady: step1.intentReady,
@@ -57,6 +57,7 @@ from app.gateway.routers import (
public_leaderboard,
public_scheduled_tasks,
recommend,
flow_extract,
recommended_questions,
roundtable_chains,
roundtable_draft_shares,
@@ -1333,6 +1334,9 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
# Roundtable-planning drafts + recommendation history at /api/roundtable-drafts
app.include_router(roundtable_drafts.router)
# Step3 流程图结构化抽取 + 校验 (SSE) at /api/flow-extract
app.include_router(flow_extract.router)
# Roundtable-planning business chains at /api/roundtable-chains
app.include_router(roundtable_chains.router)
@@ -0,0 +1,510 @@
"""流程图结构化抽取 + 校验 的 SSE 端点。
第三步「方案总结」之后,要把报告 / 各席位交付的层级结论抽成一张**流程图**(节点 + 连线)。
之前让总结智能体在同一对话线程里顺手吐 ``flow-json`` 很不稳:业务链 ``coordinator_prompt``
常写着「输出结构化结论 / 按阶段产出报告」,模型被线程上下文锚定,要么不出 flow-json、要么
结构残缺(只剩两三个节点)。
这里改成**独立的一次性模型调用**(不走 lead_agent / 不带线程历史),两段式:
① 抽取(``model.astream``):把报告/席位交付按层级抽成 flow-json,**流式**把过程吐给前端实时显示;
② 校验/修复(``model.ainvoke``):把抽取结果连同源材料再过一遍模型,补全漏层、修正连边、
去掉阶段/汇总节点,产出干净的 flow-json;后端再做容错解析 + 结构校验,返回最终图数据。
SSE 帧(``data: {json}\n\n``):
- ``{"event":"extract_delta","content":"…"}`` 抽取流式增量(前端把 JSON 实时显示出来)
- ``{"event":"extract_done"}`` 抽取结束,进入校验
- ``{"event":"validating"}`` 校验中
- ``{"event":"final","graph":{title,nodes,edges}}`` 校验后的最终图(前端据此绘制)
- ``{"event":"error","detail":"…"}``
"""
from __future__ import annotations
import json
import logging
import re
from collections.abc import AsyncIterator
from typing import Any
from fastapi import APIRouter, Body, Request
from fastapi.responses import StreamingResponse
from langchain_core.messages import HumanMessage, SystemMessage
from pydantic import BaseModel, Field
from deerflow.models.factory import create_chat_model
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/flow-extract", tags=["flow-extract"])
# ── schemas ──────────────────────────────────────────────────────────────
class SeatDelivery(BaseModel):
sender: str = ""
content: str = ""
class FlowExtractRequest(BaseModel):
# 数据来源:有报告优先用报告,否则用各席位交付。
report_md: str | None = Field(default=None)
seat_deliveries: list[SeatDelivery] = Field(default_factory=list)
# 层级骨架:优先 coordinator_prompt 里写明的层级,否则用 chain_levels。
chain_levels: list[str] = Field(default_factory=list)
coordinator_prompt: str | None = Field(default=None)
model: str | None = Field(default=None)
# Phase 2 问答修改:带上当前图 + 用户指令 → 进 edit 模式(按指令改图,其余不变)。
current_graph: dict[str, Any] | None = Field(default=None)
instruction: str | None = Field(default=None)
# ── flow-json 容错解析(Python 版,对齐前端 lib/flow-graph.ts)──────────────
_FENCE = "```flow-json"
_VALID_TYPES = {"start", "end", "stage", "task", "decision", "risk", "deliverable", "role"}
def _as_str(v: Any) -> str:
return v.strip() if isinstance(v, str) else ""
def _loose_nodes(body: str) -> list[dict[str, Any]]:
"""正则逐字段抽取节点(strict json.loads 失败时回退)。容忍裸换行 / 尾逗号。"""
# 取 "nodes" 后的 [ ... ](到与之配对的 ] 或文本结尾)。
key = body.find('"nodes"')
if key < 0:
return []
arr = body.find("[", key)
if arr < 0:
return []
# 平衡括号切出每个顶层 {…}
slices: list[str] = []
depth = 0
start = -1
in_str = False
esc = False
for i in range(arr + 1, len(body)):
ch = body[i]
if in_str:
if esc:
esc = False
elif ch == "\\":
esc = True
elif ch == '"':
in_str = False
continue
if ch == '"':
in_str = True
elif ch == "{":
if depth == 0:
start = i
depth += 1
elif ch == "}":
depth -= 1
if depth == 0 and start >= 0:
slices.append(body[start : i + 1])
start = -1
elif ch == "]" and depth == 0:
break
out: list[dict[str, Any]] = []
for sl in slices:
out.append(_loose_one(sl))
return out
def _loose_edges(body: str) -> list[dict[str, Any]]:
"""非法 JSON 回退路径下,从顶层 `"edges":[ {from,to,label} … ]` 容错抽边。"""
key = body.find('"edges"')
if key < 0:
return []
arr = body.find("[", key)
if arr < 0:
return []
depth = 0
start = -1
in_str = False
esc = False
slices: list[str] = []
for i in range(arr + 1, len(body)):
ch = body[i]
if in_str:
if esc:
esc = False
elif ch == "\\":
esc = True
elif ch == '"':
in_str = False
continue
if ch == '"':
in_str = True
elif ch == "{":
if depth == 0:
start = i
depth += 1
elif ch == "}":
depth -= 1
if depth == 0 and start >= 0:
slices.append(body[start : i + 1])
start = -1
elif ch == "]" and depth == 0:
break
def pick(slice_: str, *keys: str) -> str:
for k in keys:
m = re.search(r'"' + k + r'"\s*:\s*"([^"]*)"', slice_)
if m:
return m.group(1)
return ""
out: list[dict[str, Any]] = []
for sl in slices:
frm = pick(sl, "from", "source")
to = pick(sl, "to", "target")
if frm and to:
e: dict[str, Any] = {"from": frm, "to": to}
lbl = pick(sl, "label")
if lbl:
e["label"] = lbl
out.append(e)
return out
def _loose_one(slice_: str) -> dict[str, Any]:
def pick(*keys: str) -> str | None:
for k in keys:
m = re.search(r'"' + k + r'"\s*:\s*"((?:\\.|[^"\\])*)"', slice_)
if m:
raw = m.group(1)
try:
return json.loads('"' + raw.replace("\n", "\\n").replace("\r", "") + '"')
except Exception: # noqa: BLE001
return raw.replace("\\n", "\n").replace('\\"', '"')
return None
obj: dict[str, Any] = {}
for k, dst in (("id", "id"), ("type", "type"), ("detail", "detail")):
v = pick(k)
if v:
obj[dst] = v
label = pick("label", "name")
if label:
obj["label"] = label
group = pick("group", "level", "tier", "layer")
if group:
obj["group"] = group
fm = re.search(r'"(?:from|prev)"\s*:\s*\[([\s\S]*?)\]', slice_)
if fm:
inner = fm.group(1)
ids: list[Any] = [{"id": x} for x in re.findall(r'\{[^{}]*?"id"\s*:\s*"([^"]+)"[^{}]*?\}', inner)]
if not ids:
ids = re.findall(r'"([^"]+)"', inner)
if ids:
obj["from"] = ids
return obj
def _norm_node(raw: dict[str, Any]) -> dict[str, Any] | None:
nid = _as_str(raw.get("id"))
label = _as_str(raw.get("label")) or _as_str(raw.get("name"))
if not nid or not label:
return None
rtype = _as_str(raw.get("type")).lower()
ntype = rtype if rtype in _VALID_TYPES else "task"
node: dict[str, Any] = {"id": nid, "label": label, "type": ntype}
detail = _as_str(raw.get("detail")) or _as_str(raw.get("description"))
if detail:
node["detail"] = detail
group = _as_str(raw.get("group")) or _as_str(raw.get("level")) or _as_str(raw.get("tier")) or _as_str(raw.get("layer"))
if group:
node["group"] = group
return node
def _edges_from(raw: dict[str, Any], nid: str) -> list[dict[str, Any]]:
frm = raw.get("from") or raw.get("prev")
if not isinstance(frm, list):
return []
out: list[dict[str, Any]] = []
for e in frm:
if isinstance(e, str) and e.strip():
out.append({"from": e.strip(), "to": nid})
elif isinstance(e, dict):
fid = _as_str(e.get("id")) or _as_str(e.get("from"))
if fid:
edge = {"from": fid, "to": nid}
lbl = _as_str(e.get("label"))
if lbl:
edge["label"] = lbl
out.append(edge)
return out
def parse_flow_graph(text: str) -> dict[str, Any] | None:
"""从(可能带 fence / 非法 JSON 的)文本里解析出 ``{title, nodes, edges}``。无节点→None。"""
start = text.rfind(_FENCE)
body = text[start + len(_FENCE) :] if start >= 0 else text
close = body.find("```")
if close >= 0:
body = body[:close]
body = body.strip()
title = ""
nodes_raw: list[dict[str, Any]] = []
obj: Any = None
for candidate in (body, re.sub(r",(\s*[}\]])", r"\1", body)): # 原文 + 去尾逗号
try:
obj = json.loads(candidate)
break
except Exception: # noqa: BLE001
obj = None
if isinstance(obj, dict):
title = _as_str(obj.get("title"))
raw_list = obj.get("nodes")
nodes_raw = raw_list if isinstance(raw_list, list) else []
else:
m = re.search(r'"title"\s*:\s*"([^"]*)"', body)
if m:
title = m.group(1)
nodes_raw = _loose_nodes(body)
nodes: list[dict[str, Any]] = []
edges: list[dict[str, Any]] = []
seen: set[str] = set()
for raw in nodes_raw:
if not isinstance(raw, dict):
continue
node = _norm_node(raw)
if not node or node["id"] in seen:
continue
seen.add(node["id"])
nodes.append(node)
edges.extend(_edges_from(raw, node["id"]))
if not nodes:
return None
# 兼容**独立 `edges` 数组**:模型可能把连线写成节点上的 `from`,也可能写成顶层
# `edges:[{from,to,label}]`(或 source/target 别名)。后者若不处理,节点全无连线(bug)。
raw_edges: Any = obj.get("edges") if isinstance(obj, dict) else None
if not isinstance(raw_edges, list):
raw_edges = _loose_edges(body)
for e in raw_edges or []:
if not isinstance(e, dict):
continue
frm = _as_str(e.get("from")) or _as_str(e.get("source"))
to = _as_str(e.get("to")) or _as_str(e.get("target"))
if not frm or not to:
continue
edge: dict[str, Any] = {"from": frm, "to": to}
lbl = _as_str(e.get("label"))
if lbl:
edge["label"] = lbl
edges.append(edge)
# 只保留两端都存在的边 + 去重。
valid_edges: list[dict[str, Any]] = []
keys: set[str] = set()
for e in edges:
if e["from"] not in seen or e["to"] not in seen or e["from"] == e["to"]:
continue
k = f'{e["from"]}->{e["to"]}'
if k in keys:
continue
keys.add(k)
valid_edges.append(e)
return {"title": title, "nodes": nodes, "edges": valid_edges}
# ── prompts ────────────────────────────────────────────────────────────────
_SHAPE = (
"flow-json 严格用下面这个形状(一个 JSON 对象,nodes 是节点数组;示例仅示意结构、勿照抄内容):\n"
"```flow-json\n"
'{"title":"流程图标题","nodes":[\n'
' {"id":"n1","label":"任务简短名","type":"start","group":"任务","detail":"任务目标"},\n'
' {"id":"n2","label":"目的核心短语","type":"task","group":"目的","detail":"名称+理由要点(2-4句,单行)","from":["n1"]},\n'
' {"id":"n3","label":"行为体核心短语","type":"role","group":"行为体","detail":"名称+理由要点","from":["n2"]}\n'
"]}\n"
"```\n"
"字段:id(n1/n2…唯一)、label(≤12字,节点标题)、type(start/end/stage/task/decision/risk/deliverable/role 之一,仅决定配色)、"
"group(该节点所在层级名称,如 任务/目的/行为体/预期行为/驱动因素/叙事,必填)、"
"detail(该条目名称+理由要点,约2-4句,**写成一行**不要内含换行)、"
"from(前驱节点 id 列表,只引用前面已出现的节点)。"
)
_STRUCTURE = (
"【画法·严格遵守】"
"① 根节点 n1=本次任务本身(报告/任务总览里的任务名称),label 用任务简短名,group 填「任务」,不要用「任务」泛称当 label;"
"② 从第二层起按层级逐层展开:每层在报告/交付里通常列了多条具体条目(如 目的层有 目的1/2/3、行为体层有各自对应行为体,后面还有 预期行为·驱动因素、叙事…),"
"把每一条具体条目各画成一个节点,不要把整层合并成一个笼统节点;"
"③ 每个节点 group 填它所在层级名称(取自报告里该层级真实名称);"
"④ 连边严格按报告里的对应关系串成分支:根任务→各目的;每个目的→它对应的行为体;每个行为体→对应的预期行为·驱动因素→对应的叙事……每组各成一条自上而下的分支(有几组就几条,组间不混连);"
"⑤ 务必把每层每一条都输出成节点、不要漏层、不要只输出开头一两个节点就停(典型约13-16个节点);"
"⑥ 只画这些层级实体节点,不要把『综合分析/核心洞察/执行路线/建议/结论』及『阶段①②③④』等汇总段落/阶段名画成节点。"
)
def _material(req: FlowExtractRequest) -> str:
report = (req.report_md or "").strip()
if report:
return "===== 方案总结报告全文(数据来源,逐条抽取,勿改写)=====\n" + report + "\n===== 报告结束 ====="
seats = [s for s in req.seat_deliveries if (s.content or "").strip()]
if seats:
body = "\n\n".join(f"## 席位{i + 1}:{s.sender or '(未具名席位)'}\n{s.content.strip()}" for i, s in enumerate(seats))
return "===== 各席位交付产物(数据来源,逐条抽取,勿改写)=====\n" + body + "\n===== 交付结束 ====="
return ""
def _hierarchy_hint(req: FlowExtractRequest) -> str:
if (req.coordinator_prompt or "").strip():
return "层级顺序优先采用下面「业务链总控编排提示」里写明的层级链(如它写的 任务→目的→行为体→…):\n" + req.coordinator_prompt.strip()
levels = [s.strip() for s in req.chain_levels if s.strip()]
if levels:
return "层级顺序(自上而下):任务 → " + " → ".join(levels) + "。"
return "层级顺序:以「任务」为根,按报告/交付的逻辑结构合理分层展开。"
def _extract_messages(req: FlowExtractRequest) -> list[Any]:
sys = (
"你是「流程图结构化抽取器」。你的唯一产物是一张流程图的 flow-json 数据(节点+连线),"
"不是文字报告、不是章节、不是结论清单。只输出一个 ```flow-json 代码块,不要写任何其它正文。全程简体中文。\n"
+ _SHAPE
+ "\n"
+ _STRUCTURE
)
user = _material(req) + "\n\n" + _hierarchy_hint(req) + "\n\n请据此输出一个完整的 ```flow-json 代码块。"
return [SystemMessage(content=sys), HumanMessage(content=user)]
def _edit_messages(req: FlowExtractRequest) -> list[Any]:
"""Phase 2 问答修改:按用户指令改当前流程图(增/删/改节点或连线),其余保持不变。"""
cur = json.dumps(req.current_graph or {}, ensure_ascii=False)
sys = (
"你是「流程图编辑器」。下面给你 当前流程图(flow-json) + 数据来源 + 用户的修改指令。"
"请**按用户指令修改流程图**(可增/删/改节点、改连线、改 label/group/detail),"
"**未被指令涉及的部分保持原样**,输出**修改后的完整 flow-json**(只输出一个 ```flow-json 代码块,不要解释)。"
"节点 id 尽量沿用原有的;删节点要同时去掉与它相关的连线。全程简体中文。\n"
+ _SHAPE
)
user = (
_material(req)
+ "\n\n===== 当前流程图(flow-json,待按指令修改)=====\n"
+ cur
+ "\n===== 当前流程图结束 =====\n\n用户修改指令:\n"
+ (req.instruction or "").strip()
)
return [SystemMessage(content=sys), HumanMessage(content=user)]
def _validate_messages(req: FlowExtractRequest, extracted: str) -> list[Any]:
sys = (
"你是「流程图校验修复器」。下面给你 数据来源 和 一份已抽取的 flow-json。"
"请对照数据来源校验并修复这份 flow-json,输出**修复后**的完整 flow-json(只输出一个 ```flow-json 代码块,不要解释)。\n"
"校验要点:① 根节点是任务本身(group=任务);② 每个层级的每一条具体条目都在(没漏层、没漏条目);"
"③ 连边按对应关系成分支、无悬空边、根可达;④ 每个节点都有 group(层级名),label≤12字,detail 为单行的名称+理由要点;"
"⑤ 没有『综合分析/执行路线/结论』及『阶段编号』这类无关节点。发现问题就改对,没问题就原样规范化输出。\n"
+ _SHAPE
)
user = _material(req) + "\n\n===== 已抽取的 flow-json(待校验修复)=====\n" + extracted + "\n===== 待校验结束 ====="
return [SystemMessage(content=sys), HumanMessage(content=user)]
# ── endpoint ─────────────────────────────────────────────────────────────
def _frame(data: dict[str, Any]) -> str:
return f"data: {json.dumps(data, ensure_ascii=False, default=str)}\n\n"
def _chunk_parts(chunk: Any) -> tuple[str, str]:
"""从流式 chunk 取 (正文增量, 思考增量)。思考走 additional_kwargs.reasoning_content
(DeepSeek/兼容 OpenAI 推理模型的单列字段),正文走 content。"""
content = chunk.content if isinstance(getattr(chunk, "content", None), str) else ""
ak = getattr(chunk, "additional_kwargs", None) or {}
reasoning = ""
if isinstance(ak, dict):
r = ak.get("reasoning_content") or ak.get("reasoning")
if isinstance(r, str):
reasoning = r
return content, reasoning
@router.post("/stream")
async def extract_flow_stream(request: Request, payload: FlowExtractRequest = Body(...)) -> StreamingResponse:
"""流式抽取流程图 JSON + 流式校验修复(抽取/校验过程含思考都实时吐前端)→ 返回最终结构化图。"""
async def gen() -> AsyncIterator[str]:
try:
# 带 instruction + current_graph → Phase 2 问答修改(按指令改当前图);否则首次抽取。
is_edit = bool((payload.instruction or "").strip()) and payload.current_graph is not None
if not _material(payload).strip() and not is_edit:
yield _frame({"event": "error", "detail": "缺少数据来源(报告或各席位交付均为空)"})
return
# ① 抽取 / 编辑(正文 + 思考(reasoning) 拼成 `<think>…</think>正文` 增量逐帧吐,前端实时显示)
round1_messages = _edit_messages(payload) if is_edit else _extract_messages(payload)
model = create_chat_model(name=payload.model or None, thinking_enabled=True)
extracted = ""
think_open = False
async for chunk in model.astream(round1_messages):
c, r = _chunk_parts(chunk)
out = ""
if r:
if not think_open:
out += "<think>"
think_open = True
out += r
if c:
if think_open:
out += "</think>"
think_open = False
out += c
extracted += c
if out:
yield _frame({"event": "extract_delta", "content": out})
if think_open:
yield _frame({"event": "extract_delta", "content": "</think>"})
yield _frame({"event": "extract_done"})
# ② 校验 / 修复(同样流式 + 思考实时显示)
yield _frame({"event": "validating"})
validated = ""
graph: dict[str, Any] | None = None
try:
vmodel = create_chat_model(name=payload.model or None, thinking_enabled=True)
think_open = False
async for chunk in vmodel.astream(_validate_messages(payload, extracted)):
c, r = _chunk_parts(chunk)
out = ""
if r:
if not think_open:
out += "<think>"
think_open = True
out += r
if c:
if think_open:
out += "</think>"
think_open = False
out += c
validated += c
if out:
yield _frame({"event": "validate_delta", "content": out})
if think_open:
yield _frame({"event": "validate_delta", "content": "</think>"})
graph = parse_flow_graph(validated)
except Exception: # noqa: BLE001 — 校验失败不致命,回退抽取结果
logger.warning("flow-extract validation failed, fall back to extraction", exc_info=True)
graph = None
if graph is None:
graph = parse_flow_graph(extracted)
if graph is None:
yield _frame({"event": "error", "detail": "未能从模型输出解析出流程图节点"})
return
yield _frame({"event": "final", "graph": graph})
except Exception as exc: # noqa: BLE001
logger.exception("flow-extract stream failed")
yield _frame({"event": "error", "detail": str(exc)})
return StreamingResponse(
gen(),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no", "Connection": "keep-alive"},
)
+12 -12
View File
@@ -10,18 +10,18 @@ gateway:
port: 8001
models:
# - name: deepseek-chat
# display_name: DeepSeek Chat
# use: langchain_openai:ChatOpenAI
# model: deepseek-chat
# api_key: sk-e1c22b7931dc446f8b67e57878edd105
# base_url: https://api.deepseek.com
# request_timeout: 600.0
# max_retries: 2
# max_tokens: 4096
# temperature: 1.0
# supports_vision: false
# supports_thinking: true
- name: deepseek-chat
display_name: DeepSeek Chat
use: langchain_openai:ChatOpenAI
model: deepseek-chat
api_key: sk-35345154983047aba9e12faa86c5f62c
base_url: https://api.deepseek.com
request_timeout: 600.0
max_retries: 2
max_tokens: 4096
temperature: 1.0
supports_vision: false
supports_thinking: true
# - name: zai-org/GLM-5-FP8
# display_name: zai-org/GLM-5-FP8